package realtime.func;

import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import realtime.bean.TableProcess;
import realtime.common.GmallConfig;
import realtime.util.DruidDSUtil;
import realtime.util.JdbcUtil;
import realtime.util.JedisUtil;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @author MengX
 * @create 2023/3/24 14:30:41
 */
public class DimMyprocessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    private final MapStateDescriptor<String, TableProcess> broadcastState;
    private HashMap<String, TableProcess> tableProcessHashMap;

    public DimMyprocessFunction(MapStateDescriptor<String, TableProcess> broadcastState) {
        this.broadcastState = broadcastState;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        tableProcessHashMap = new HashMap<>();
        Connection connet = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/gmall_config?" + "user=root&password=123456&useUnicode=true&" +
                "characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        String sql = "select * from table_process where sink_type = 'dim'";
        List<TableProcess> tableProcesses = JdbcUtil.queryList(connet, sql, TableProcess.class, true);
        for (TableProcess tableProcess : tableProcesses) {
            String sourceTable = tableProcess.getSourceTable();
            tableProcessHashMap.put(sourceTable,tableProcess);
        }
    }

    //FlinkCDC：
    //{"before":null,"after":{"source_table":"base_category3","sink_table":"dim_base_category3","sink_columns":"id,name,category2_id","sink_pk":"id","sink_extend":null},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1669162876406,"snapshot":"false","db":"gmall-220623-config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1669162876406,"transaction":null}
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(this.broadcastState);
        JSONObject flinCdc = JSONObject.parseObject(value);
        String op = flinCdc.getString("op");
        if ("d".equals(op)){
            String outKey = flinCdc.getJSONObject("before").getString("source_table");
            String sinkType = flinCdc.getJSONObject("before").getString("sink_type");
            if ("dim".equals(sinkType)){
                broadcastState.remove(outKey);
                tableProcessHashMap.remove(outKey);
            }
        }else {
            JSONObject after = flinCdc.getJSONObject("after");
            if ("dim".equals(after.getString("sink_type"))){
                String source_table = after.getString("source_table");
                TableProcess tableProcess = JSONObject.parseObject(flinCdc.getString("after"), TableProcess.class);
                //建立phionx表
                createTble(after);
                broadcastState.put(source_table,tableProcess);
            }
        }
    }
    //建表:create table if not exists db.tn
    // (id varchar primary key ,name varchar ,sex varchar) xxx
    private void createTble(JSONObject after) throws SQLException {
        String sinkPk = after.getString("sink_pk");
        String sinkExtend = after.getString("sink_extend");
        if (sinkPk == null){
            sinkPk = "id";
        }
        if (sinkExtend == null){
            sinkExtend = "";
        }
        StringBuilder createSql = new StringBuilder();
        String sinkTable = after.getString("sink_table");
        createSql.append("create table if not exists " + GmallConfig.HBASE_SCHEMA + ".").append(sinkTable).append(" (");
        String sinkColumns = after.getString("sink_columns");
        String[] split = sinkColumns.split(",");
        for (int i = 0; i < split.length; i++) {
            String s = split[i];
            createSql.append(s).append(" varchar");
            if (sinkPk.equals(s)){
                createSql.append(" primary key ");
            }
            if (i < split.length -1){
                createSql.append(",");
            }else createSql.append(")").append(sinkExtend);
        }
        System.out.println("建表语句："+ createSql);
        //执行建表语句
        DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
        String sql = createSql.toString();
        try{
            PreparedStatement preparedStatement = phoenixConn.prepareStatement(sql);
            preparedStatement.execute();
            phoenixConn.close();
        }catch (Exception e){
            System.out.println("建表失败："+ sql);
        }
    }
    //Maxwell-topic_db
    //{"database":"gmall-220623-flink","table":"comment_info","type":"insert","ts":1669162958,"xid":1111,"xoffset":13941,"data":{"id":1595211185799847960,"user_id":119,"nick_name":null,"head_img":null,"sku_id":31,"spu_id":10,"order_id":987,"appraise":"1204","comment_txt":"评论内容：48384811984748167197482849234338563286217912223261","create_time":"2022-08-02 08:22:38","operate_time":null}}
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(this.broadcastState);
        String table = value.getString("table");
        TableProcess tableProcess = broadcastState.get(table);
        TableProcess tableProcess1 = tableProcessHashMap.get(table);
        String type = value.getString("type");
        if ("insert".equals(type) || "update".equals(type) || "bootstrap-insert".equals(type)){
            if (tableProcess != null || tableProcess1 != null){
                if (tableProcess == null){
                    tableProcess = tableProcess1;
                }
                JSONObject data = value.getJSONObject("data");
                if ("update".equals(type)){
                    Jedis jedis = JedisUtil.getJedis();
                    String key = "DIM:"+tableProcess.getSinkTable()+data.getString("id");
                    jedis.setex(key,24*3600,data.toString());
                    jedis.close();
                }
                columnsFlit(data,tableProcess);
                String sinkTable = tableProcess.getSinkTable();
                data.put("sink_table",sinkTable);
                out.collect(data);
            }
        }
    }

    private void columnsFlit(JSONObject data, TableProcess tableProcess) {
        String sinkColumns = tableProcess.getSinkColumns();
        String[] split = sinkColumns.split(",");
        List<String> strings = Arrays.asList(split);
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            String key = next.getKey();
            if (!strings.contains(key)){
                iterator.remove();
            }
        }
    }
}
